package com.dayuanit.dy17.mix.proxy;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.Socket;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;

public class PayInvocationHandler implements InvocationHandler {

    //服务提供者列表
    private static final List<IntenetAddress> address = new CopyOnWriteArrayList<>();

    private static CuratorFramework zkClient;
    private static final String zkServerHost = "192.168.2.139:2181";
    private static final int sessionTimeOut = 30 * 1000;
    private static final int connectionTimeOut = 30 * 1000;

    static {

        initZookeeper();

        try {
            //监听服务节点的变化
            watchProviderNode();
            //加载服务列表信息
            loadProviderInfos();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 初始化zookeeper连接
     */
    private static void initZookeeper() {
        //重试策略 最多重试三次，每次间隔1秒钟
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkClient = CuratorFrameworkFactory.newClient(zkServerHost, sessionTimeOut, connectionTimeOut, retryPolicy);
        //发起和zk服务器连接请求
        zkClient.start();
    }

    /**
     * 监控服务节点的变化
     */
    private static void watchProviderNode() throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, "/dayuanit/17/rpc/provider", true);
        pathChildrenCache.start();

        pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                System.out.println("------ 服务节点发生了变化 ------");
                loadProviderInfos();
            }
        });

    }

    private static void loadProviderInfos() throws Exception {
        //先清空节点信息
        address.clear();

        List<String> providers = zkClient.getChildren().forPath("/dayuanit/17/rpc/provider");
        for (String nodeName : providers) {
            byte[] bytes = zkClient.getData().forPath("/dayuanit/17/rpc/provider/" + nodeName);

            //com.dayuanit.dy17.mix.service.impl.PayServiceImpl|127.0.0.1|10000
            String providerInfo = new String(bytes);
            System.out.println(providerInfo);

            //由于|符号是正则表达式的关键字，所以要将|进行转义
            String[] split = providerInfo.split("\\|");
            //把服务提供者信息解析后 封装成对象
            IntenetAddress intenetAddress = new IntenetAddress(split[1], Integer.parseInt(split[2]), split[0]);
            address.add(intenetAddress);
        }
    }

    //负载均衡策略 随机
    public static IntenetAddress loadBalance() {
        Random random = new Random();
        int index = random.nextInt(address.size());
        System.out.println("负载均衡Index=" + index);
        return address.get(index);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //通过RPC的方式 调用provider项目中的pay方法
        //发送以下4个信息：
        //1.类名(包名+类名)
        //2.方法名
        //3.参数类型
        //4.参数值

        //通过负责均衡 得到一个服务器的信息
        IntenetAddress intenetAddress = loadBalance();

        //跟服务器提供者建立TCP/IP连接
        Socket socket = new Socket(intenetAddress.host, intenetAddress.port);

        //JDK序列化  通过JDK序列化 将信息发送给服务提供者
        ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
        //类名
        oos.writeUTF(intenetAddress.clazzName);
        //方法名
        oos.writeUTF(method.getName());
        //参数类型
        oos.writeObject(method.getParameterTypes());
        //参数值
        oos.writeObject(args);

        oos.flush();

        //读取服务提供者返回的信息
        ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());
        Object result = ois.readObject();
        return result;
    }




    static class IntenetAddress {
        public String host;
        public int port;
        public String clazzName;

        public IntenetAddress(String host, int port, String clazzName) {
            this.host = host;
            this.port = port;
            this.clazzName = clazzName;
        }
    }
}
